package com.event.ingestion.kafka.streamer;

import com.event.ingestion.kafka.BaseStreamer;

import java.util.ArrayList;
import java.util.List;

/**
 * @author ZzHh
 * @Classname UserFriendsStreamer
 * @Description TODO
 * @Date 2020/6/17 11:35
 * @Create by IntelliJ IDEA
 **/

/**
 * UserFriend数据变换
 */
public class UserFriendsStreamer extends BaseStreamer {
    @Override
    protected String getApplicationId() {
        return "user-friends-streamming";
    }

    @Override
    protected String getSourceTopic() {
        return "user_friends_raw";
    }

    @Override
    protected String getTargetTopic() {
        return "user_friends";
    }

    //数据变换业务代码实现
    @Override
    protected List<String[]> transform(String[] fields) {
        List<String[]> results = new ArrayList<String[]>();
        //fields [123123,123123 123754 12512 5432 213]
        //结果为:[123123,123123][123123,123754]
        String user_id = fields[0];
        //friends
        String [] friends = fields[1].split(" ");
        //check
        if (friends != null && friends.length > 0) {
            //循环
            for (String friend_id :friends) {
                //add
                results.add(new String[]{user_id,friend_id});
            }
        }
        return results;
    }

    @Override
    protected Boolean isHeader(String[] fields) {
        //check
        return (isValid(fields) && fields[0].equals("user") && fields[1].equals("friends"));
    }

    @Override
    protected Boolean isValid(String[] fields) {
        return fields.length > 1;
    }
}